package com.bjzh.bjzh.util;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;


@Configuration
@IntegrationComponentScan
public class MqttReader {
    @Value("${spring.mqtt.username}")
    private String username;
    @Value("${spring.mqtt.password}")
    private String password;
    @Value("${spring.mqtt.serverURIs}")
    private String hostUrl;
    @Value("${spring.mqtt.client.id}")
    private String clientId;
    @Value("${spring.mqtt.client.listenClientId}")
    private String listenClientId;
    @Value("${spring.mqtt.client.topic}")
    private String defaultTopic;
    @Value("${spring.mqtt.timeout}")
    private int completionTimeout;
    @Value("${spring.mqtt.keepalive}")
    private int keepalive;


    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setCleanSession(true);
        factory.setConnectionTimeout(completionTimeout);
        factory.setUserName(username);
        factory.setPassword(password);
        factory.setServerURIs(hostUrl);
        factory.setKeepAliveInterval(keepalive);
        return factory;
    }


    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {

        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(listenClientId, mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(defaultTopic);
        messageHandler.setDefaultRetained(false);
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel mqttInboundChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT消息处理器（消费者）
     *
     * @return {@link org.springframework.messaging.MessageHandler}
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInboundChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) {


                JSONObject jsonObject = new JSONObject(message.getPayload());
                String event = jsonObject.get("event").toString();
                if (ObjectUtil.isNull(event)){
                    event=jsonObject.get("topic").toString();
                }
                System.err.println(jsonObject);
                /**
                switch (event) {
                    case "heartbeat":
                        String macNo = jsonObject.get("macNo").toString();
                        redisUtils.set(macNo+"_online",true);
                        msgWriter.sendToMqtt("gdjbhsx_push/" + macNo, 0, new JSONObject().set("topic", "heartbeat_back").set("macno", macNo).toString());
                        break;
                    case "openback":
                        String status = jsonObject.getStr("status");
                        String deviceNo = jsonObject.getStr("deviceNo");
                        String type_num = jsonObject.getStr("type_num");
                        String type_id = jsonObject.getStr("type_id");
                        String oid = jsonObject.getStr("oid");


                        break;
                    case "operateLog":
                        break;
                    default:
                        break;
                }*/
            }
        };
    }

    @Bean
    public MessageProducer mqttInbound() {
        // 可以同时消费（订阅）多个Topic
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(),
                "pub/#");
        adapter.setCompletionTimeout(completionTimeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(0);
        // 设置订阅通道
        adapter.setOutputChannel(mqttInboundChannel());
        return adapter;
    }
}
